package com.learn.kafka.interceptor;


import com.learn.kafka.config.ProducerProperties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.ArrayList;
import java.util.List;

/**
 * 带拦截器的生产者，继承 ProducerProperties 复用配置
 * @author:         MI
 * @email:          448341911@qq.com
 * @createTime:     2025/8/8 22:51
 * @updateUser:     MI
 * @updateTime:     2025/8/8 22:51
 * @updateRemark:   修改内容
 * @version:        1.0
 */
public class InterceptorProducer extends ProducerProperties {
    public static void main(String[] args) {
        // 创建带拦截器的生产者实例
        InterceptorProducer producer = new InterceptorProducer();
        producer.startMessages();
    }

    // 构造方法：初始化拦截器配置
    public InterceptorProducer() {
        super(); // 调用父类构造方法，加载基础配置
        initInterceptorConfig();
    }

    // 初始化拦截器配置
    private void initInterceptorConfig() {
        // 添加自定义拦截器
        List<String> interceptors = new ArrayList<>();
        interceptors.add("com.learn.kafka.interceptor.ChargingInterceptor");
        super.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, String.join(",", interceptors));
    }

    // 发送消息的方法
    private void startMessages() {
        // 调用父类方法创建生产者，消除重复代码
        Producer<Integer, String> producer = super.createProducer();

        // 发送消息
        producer.send(new ProducerRecord<Integer,String>("mytopic",1,"1"));
        producer.send(new ProducerRecord<Integer,String>("mytopic",2,"2"));
        // 关闭生产者
        producer.close();
    }

}
